package com.cxdr.awsips.distribute.processor;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.cxdr.awsips.common.db.DBHelper;
import com.cxdr.awsips.common.logger.LogFactory;
import com.cxdr.awsips.common.thread.BaseProcessor;
import com.cxdr.awsips.distribute.bean.ReportFileLog;
import com.cxdr.awsips.distribute.config.SystemConfig;
import com.cxdr.awsips.distribute.pool.BaseQueue;

/**
 * 存储到报日志线程
 * 
 * @author pugang
 * 
 */
public class StoreReportProccessor extends BaseProcessor {
	public static final BaseQueue<ReportFileLog> ReportFileLogQueue = new BaseQueue<ReportFileLog>();
	private static Logger logger = LogFactory.makeNewLogger(StoreReportProccessor.class);
	private static final int BATCH_SIZE = Integer.parseInt(SystemConfig.get("num.batchsize"));
	private StoreThread[] ths = new StoreThread[Integer.parseInt(SystemConfig.get("num.reportlog-threads"))];
	private ExecutorService executor = Executors.newFixedThreadPool(ths.length);
	private static final String insertSql = "INSERT INTO t_report_file_log(id,partition_month,create_time,file_type,reach_time,file_name,station_id,obser_time,file_size,is_update,update_count,create_minute,IS_POINT,timely_flag)"
			+ " VALUES     ( SEQ_T_REPORT_FILE_LOG.NEXTVAL ,to_char(sysdate,'yyyy-mm'),sysdate,?,?,?,?,?,?,?,?,?,?,'0')";
	// private static final String selectSql =
	// "select * from t_report_file_log where file_name= ?";
	private static final String updateSql = "UPDATE t_report_file_log  SET    obser_time = ?,is_update = 'Y',update_count = ?  WHERE  file_name = ?";

	private class StoreThread extends BaseProcessor {
		private Connection connection;

		StoreThread() {
			try {
				connection = DBHelper.getConnection();
				connection.setAutoCommit(false);
			} catch (SQLException e) {
				this.finish();
				logger.error("GetConnection Error", e);
				logger.error("StoreThread will not real running ....");
//				finish();
				e.printStackTrace();
			}

		}

		@Override
		public void run() {
			logger.info("Start StoreReportProccessor : " + this.getId());
			CallableStatement insertPs = null;
			// PreparedStatement selectPs = null;
			CallableStatement updatePs = null;
			try {
				insertPs = connection.prepareCall("{call insert_report(?,?,?,?,?,?,?,?,?,?,?)}");//connection.prepareStatement(insertSql);
				updatePs = connection.prepareCall("{call update_report(?,?,?,?)}"); //connection.prepareStatement(updateSql);
				insertPs.registerOutParameter(11, Types.INTEGER);
				updatePs.registerOutParameter(4, Types.INTEGER);
				// selectPs = connection.prepareStatement(selectSql);
			} catch (SQLException e) {
				e.printStackTrace();
				logger.error("到报入库线程初始化错误",e);
				this.finish();
			}
//			StringBuilder params = new StringBuilder();
			while (!isFinish()) {
				int count = 0;
				while (!ReportFileLogQueue.isEmpty() && count < BATCH_SIZE) {
					try {
						ReportFileLog log = ReportFileLogQueue.take();
						if (log == null) {
							continue;
						}
						// 如果是更新文件则查询原来的数据，然后更新之
						if (log.isIs_update()) {
							logger.info("更新到报报文 " + log.getFile_name());
							updatePs.setDate(1, new java.sql.Date(log
									.getObser_time().getTime()));
							updatePs.setInt(2, log.getUpdate_count());
							updatePs.setString(3, log.getFile_name());
							updatePs.execute();
//							logger.updatePs.getInt(11);
//							updatePs.addBatch();
						} else {
							// 写入分发日志
							logger.info("插入到报报文 " + log.getFile_name()+"||queue size:"+ReportFileLogQueue.size());
							insertPs.setString(1, log.getFile_type());
//							params.append(log.getFile_type());
							if(log.getReach_time()!=null){
								insertPs.setTimestamp(2, new java.sql.Timestamp(log
										.getReach_time().getTime()));
							}else{
								insertPs.setTimestamp(2, null);
							}
//							params.append(",").append( new java.sql.Timestamp(log
//										.getReach_time().getTime()));
							insertPs.setString(3, log.getFile_name());
							insertPs.setString(4, log.getStation_ID());
//							params.append(",").append(log.getFile_name()).append(",").append(log.getStation_ID());
							if(log.getObser_time()!=null){
								insertPs.setTimestamp(5, new java.sql.Timestamp(log
										.getObser_time().getTime()));
							}else{
								insertPs.setTimestamp(5, null);
							}
//							params.append(",").append( new java.sql.Timestamp(log
//										.getObser_time().getTime()));
							insertPs.setLong(6, log.getFile_size());
							insertPs.setString(7,
									log.isIs_update() == true ? "1" : "0");
							insertPs.setInt(8, log.getUpdate_count());
//							params.append(",").append(log.getFile_size()).append(", '0'").append(",").append(log.getUpdate_count());
							insertPs.setString(9, ""+log.getObser_time().getMinutes());
//							params.append(",").append(log.getObser_time().getMinutes());
							String ispoint = "0";
							ispoint =log.getObser_time().getMinutes() == 0 ? "1":"0";
							insertPs.setString(10,ispoint);
//							insertPs.setInt(11, 5);
//							params.append(",").append(ispoint).append(",0");
							insertPs.execute();
							logger.info("插入结果："+insertPs.getInt(11));
//							insertPs.addBatch();
//							logger.info("insertPs.addBatch();");
//							logger.info(params);
							connection.commit();
						}
					} catch (Exception e) {
						e.printStackTrace();
						logger.error("插入到报日志错误",e);
					}
					count++;
				}
				if (count == 0) {
					continue;
				}
				try {
					logger.info("提交插入到报");
//					insertPs.executeBatch();
					connection.commit();
//					insertPs.clearBatch();
				} catch (SQLException e) {
					logger.error("插入到报错误", e);
					e.printStackTrace();
					try {
						connection.rollback();
					} catch (SQLException e1) {
						e1.printStackTrace();
						logger.error("插入到报日志错误",e);
						logger.error("数据回滚");
					}
				}
				try {
					logger.info("提交更新到报");
//					updatePs.executeBatch();
					connection.commit();
					updatePs.clearBatch();
				} catch (SQLException e) {
					logger.error("更新到报错误", e);
					e.printStackTrace();
					try {
						connection.rollback();
					} catch (SQLException e1) {
						e1.printStackTrace();
						logger.error("更新到报日志错误",e);
					}

				}
				if (ReportFileLogQueue.isEmpty()) {
					// finish();
					// StoreReportProccessor.this.finish();
					// logger.info("结束入库线程");
				}
			}
		}
	}

	public StoreReportProccessor() {
		for (int i = 0; i < ths.length; i++) {
			ths[i] = new StoreThread();
		}
	}

	@Override
	public void run() {
		logger.info("开始到报入库线程");
		for (StoreThread thread : ths) {
			executor.submit(thread);
		}
	}

}
